package cn.com.ytst.data.integration.core.channel;

import com.google.common.eventbus.EventBus;
import lombok.extern.slf4j.Slf4j;

import java.util.Collection;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName EventPipleChannel
 * @Deacription 基于事件通知模式
 * @Author Administrator
 * @Date 2020/5/10 0010 20:50
 * @Version 1.0
 **/
@Slf4j
public class EventPipleChannel implements PipleChannel,EventListener{

    private EventBus eventBus;

    private SynchronousQueue<String> synchronousQueue;

    public EventPipleChannel(){
        eventBus = new EventBus("default");
        eventBus.register(this);
        synchronousQueue = new SynchronousQueue<>();
    }
    @Override
    public void listener(String record) {
        try {
            synchronousQueue.put(record);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
       
    }

    @Override
    public void doPush(String record) {
        eventBus.post(record);
    }

    @Override
    public void doPushAll(Collection<String> records) {

    }

    @Override
    public String doPull() {

        try {
            return synchronousQueue.poll(2, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
            return null;
        }

    }

    @Override
    public boolean isEmpty() {
        return synchronousQueue.isEmpty();
    }

    @Override
    public int size() {
        return 0;
    }

    @Override
    public void readflush() {

    }
}
